package com.resoft.bigdata.p2p.service.api.thrift2imp;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.thrift2.generated.TColumnValue;
import org.apache.hadoop.hbase.thrift2.generated.THBaseService;
import org.apache.hadoop.hbase.thrift2.generated.TResult;
import org.apache.hadoop.hbase.thrift2.generated.TScan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.resoft.bigdata.p2p.record.RecordInterface;
import com.resoft.bigdata.util.DateUtil;

public abstract class AbsThrift2Client {
	public static final byte[] ZERO_POSTFIX = new byte[]{'0'};
	private static final Logger logger = LoggerFactory.getLogger(AbsThrift2Client.class);
	
	protected TTransport transport;
	
	protected THBaseService.Iface client;
	
	protected final int DEFAULT_MAX_RESULT_SIZE = 10000;
	
	public AbsThrift2Client(String host, int port){
		transport = new TSocket(host, port);
		TProtocol protocol = new TBinaryProtocol(transport);
		client = new THBaseService.Client(protocol);
	}
	
	public void openTransport() throws TTransportException{
		transport.open();
	}
	
	public void closeTransport(){
		transport.close();
	}
	
	public byte[] createNextRowKeyForOrgCodeDate(String orgCode, String date, byte[] lastRow){
		byte[] startRow = null;
		if(lastRow ==  null){
			startRow = Bytes.add(Bytes.toBytes(orgCode.hashCode()), Bytes.toBytes(DateUtil.reverseDate(date)));
		}else{
			startRow = Bytes.add(lastRow, ZERO_POSTFIX);
		}
		
		return startRow;
	}
	
	public List<RecordInterface> getRecordsByOrgCodeAndDates(String orgCode,
			String beginDate, String endDate, long pageSize, byte[] lastRow, Class<? extends RecordInterface>clazz) throws InstantiationException, IllegalAccessException {
		List<RecordInterface> records = new ArrayList<RecordInterface>();
		RecordInterface recordType = clazz.newInstance();
		
		String table = recordType.getTableName(); //LoanContract.TABLE_NAME;
		String startRowDate = endDate;
		String stopRowDate = DateUtil.addSomeDays(beginDate, -1);
		byte[] startRow = Bytes.toBytes(0);
		byte[] stopRow = createNextRowKeyForOrgCodeDate(orgCode, stopRowDate, lastRow);;
		if(pageSize > DEFAULT_MAX_RESULT_SIZE || pageSize <1){
			pageSize = DEFAULT_MAX_RESULT_SIZE;
		}
		
		try {
			transport.open();
		} catch (TTransportException e) {
			logger.error("Thrift2 Client open transport failed:", e.getMessage());
			e.printStackTrace();
			return records;
		}
		int totalNumber = 0;
//		LoanContract contractType = new LoanContract();
		while(true){
			TScan scan = new TScan();
			startRow = createNextRowKeyForOrgCodeDate(orgCode, startRowDate, lastRow);
			scan.setStartRow(startRow);
			scan.setStopRow(stopRow);
			scan.setMaxVersions(1);
			
			int localNumber = 0;
			List<TResult> results;
			try {
				results = client.getScannerResults(ByteBuffer.wrap(table.getBytes()), scan, (int)pageSize);
				for (TResult tResult : results) {
					++localNumber;
					lastRow = tResult.getRow();
					RecordInterface record = clazz.newInstance();
					record.setStoredRowKey(lastRow);
					for (TColumnValue columnValue : tResult.getColumnValues()) {
//						System.out.println(Bytes.toString(columnValue.getQualifier())+" = " +Bytes.toString(columnValue.getValue()));
						record.setValue(columnValue.getQualifier(), columnValue.getValue());
					}
					records.add(record);
//					System.out.println();
				}
	
				totalNumber += localNumber;
				if(localNumber == 0){
					System.out.println("BREAK.................");
					break;
				}
			} catch (TException e) {
				logger.error("Fetch Data failed! Message:" + e.getMessage());
				e.printStackTrace();
				transport.close();
				break;
			}
//			System.err.println("Total number:" + totalNumber + ", local number:" + localNumber +  ", row Key:" + Bytes.toHex(lastRow));
		} //~~while

		logger.debug("Get {} rows", totalNumber);
		System.err.println("Total number:" + totalNumber );
		
		transport.close();
		
		return records;
	}
	
}
